package com.one.api.push.msg.consumer;

import com.one.api.common.msg.constant.MsgConstant;
import com.one.api.common.msg.entity.body.EmailMsgBody;
import com.one.api.push.email.sender.AliEmailUtil;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Component
public class EmailKafkaReceiver {
    private Logger logger=LoggerFactory.getLogger(getClass());

    @KafkaListener(topics = {MsgConstant.emailKafkaTopic})
    public void emailKafkaListen(ConsumerRecord<String, String> record){
        Optional<String> optionalMsg=Optional.ofNullable(record.value());
        if (optionalMsg.isPresent()){
            String msg=optionalMsg.get();
            logger.debug("收到邮件消息：\n"+msg);
            JSONObject jsonObject=JSONObject.fromObject(msg);
            EmailMsgBody emailMsgBody=(EmailMsgBody)(JSONObject.toBean(jsonObject.getJSONObject("body"), EmailMsgBody.class));
            AliEmailUtil.sendSingleEmail(emailMsgBody);
        }
    }
}
